package org.cainiao.scenario.highconcurrency;

import jakarta.annotation.PostConstruct;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.cainiao.scenario.highconcurrency.mapper.OrderMapper;
import org.cainiao.scenario.highconcurrency.service.StockService;
import org.redisson.api.RBloomFilter;
import org.redisson.api.RRateLimiter;
import org.springframework.dao.DuplicateKeyException;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;

import java.util.Collections;

// 秒杀高并发场景伪代码
@RequiredArgsConstructor
@Service
@Slf4j
public class SeckillService {

    private final RocketMQTemplate rocketMQTemplate;
    private final RedisTemplate<String, Integer> redisTemplate;
    private final OrderMapper orderMapper;
    private final StockService stockService;
    private final RBloomFilter<String> userIdBloomFilter;
    private final RRateLimiter seckillRateLimiter;

    // 1. 库存预热 + Redis原子操作
    @PostConstruct
    public void initStock() {
        redisTemplate.opsForValue().set("seckill:stock:1001", 1000);
    }

    // 2. 请求拦截层（网关层）
    public boolean checkRequest(String userId, String itemId) {
        // 布隆过滤器过滤无效请求
        if (!userIdBloomFilter.contains(userId)) {
            return false;
        }
        // 限流（令牌桶算法）
        return seckillRateLimiter.tryAcquire();
    }

    // 3. 库存扣减（Lua 脚本保证原子性）
    public int deductStock(String itemId) {
        String script = "if redis.call('get', KEYS[1]) > '0' then " +
            "redis.call('decr', KEYS[1]) return 1 else return 0 end";
        return redisTemplate.execute(new DefaultRedisScript<>(script, Integer.class),
            Collections.singletonList("seckill:stock:" + itemId), "");
    }

    // 4.1 库存扣减成功后发送 MQ 消息（异步削峰-生产者）
    public void sendSeckillMessage(String userId, String itemId) {
        SeckillMessage message = new SeckillMessage(userId, itemId);
        // 发送延迟消息（可选：防止下游消费过快）
        rocketMQTemplate.syncSend(
            "seckill_topic:order_tag", // topic:tag
            MessageBuilder.withPayload(message).build(),
            3000 // 发送超时时间
        );
    }

    // 4.2 订单处理（异步削峰-消费者）
    @RocketMQMessageListener(
        topic = "seckill_topic",
        selectorExpression = "order_tag", // 选择tag
        consumerGroup = "seckill_order_group"
    )
    public class OrderListener implements RocketMQListener<SeckillMessage> {
        @Override
        public void onMessage(SeckillMessage message) {
            try {
                // 数据库唯一索引防重复
                orderMapper.insertOrder(message.getUserId(), message.getItemId());
                // 同步库存数据库（最终一致性）
                stockService.reduceDBStock(message.getItemId());
            } catch (DuplicateKeyException e) {
                // 幂等性处理：已存在的订单直接跳过
                log.warn("重复订单: {}", message.getOrderId());
            }
        }
    }
}
